package com.java.spark


import org.apache.log4j.{Level, Logger}

import scala.Iterator
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession

object mapPartition {
  Logger.getLogger("org").setLevel(Level.ERROR)
  def sumOfEveryPartition(input: Iterator[Int]): Int = {
    var total = 0
    input.foreach { elem =>
      total += elem
    }
    total
  }
  def main(args: Array[String]) {
//    val conf = new SparkConf().setAppName("Spark Rdd Test").setMaster("local");
//    val spark = new SparkContext(conf)
//    spark.setLogLevel("WARN")

    val spark1 = SparkSession.builder.master("local").appName("Word Count").getOrCreate();
    val spark = spark1.sparkContext
    val input = spark.parallelize(List(1, 2, 3, 4, 5, 6), 2)//RDD有6个元素，分成2个partition
    val result = input.mapPartitions(
      partition => Iterator(sumOfEveryPartition(partition)))//partition是传入的参数，是个list，要求返回也是list，即Iterator(sumOfEveryPartition(partition))
    result.collect().foreach {
      println(_)
//      # 6 15,分区计算和
    }

    val a = spark.parallelize(1 to 4, 2)
    val b = a.flatMap(x => 1 to x)//每个元素扩展
    val cc = b.collect()
    println(cc.toSeq)
    println(b.collect().toString)
    println(cc.foreach(print(_,",")))


    val c = spark.parallelize(List((1,2),(3,4),(5,6)))
    val d = c.flatMapValues(x=>1 to x)
    d.collect.foreach(println(_))
    /*结果
    (1,1)
    (1,2)
    (3,1)
    (3,2)
    (3,3)
    (3,4)
    (5,1)
    (5,2)
    (5,3)
    (5,4)
    (5,5)
    (5,6)
    */

    val list = List(("mobin",22),("kpop",20),("lufei",23))
    val rdd = spark.parallelize(list)
    val mapValuesRDD = rdd.flatMapValues(x => Seq(x,"male",3))
    mapValuesRDD.foreach(println)

    val df = spark1.createDataFrame(Seq(
      ("ming", 20, 15552211521L),
      ("hong", 19, 13287994007L),
      ("zhi", 21, 15552211523L)
    )).toDF("name", "age", "phone")

    df.show()



//
//    　　一、输入分区与输出分区一对一型
//
//    　　　　1、map算子
//
    println("map")



//    　　　　2、flatMap算子
//
    println("flatMap算子")

//    　　　　3、mapPartitions算子

    println("mapPartitions算子")
//
//    　　　　4、glom算子
//
    println("glom算子")


//    　　二、输入分区与输出分区多对一型　
//
//    　　　　5、union算子
    println("union算子")
//
//    　　　　6、cartesian算子
//
    println("cartesian算子")
//    　　三、输入分区与输出分区多对多型
//
//    　　　　7、grouBy算子
    println("grouBy算子")
//    　　四、输出分区为输入分区子集型
//
//    　　　　8、filter算子
    println("filter算子")
//    　　　　9、distinct算子
    println("distinct算子")
//    　　　　10、subtract算子
    println("csubtract算子")
//    　　　　11、sample算子
    println("sample算子")
//    　　 12、takeSample算子
    println("takeSample算子")
//    　　五、Cache型
//
//    　　　　13、cache算子　　
    println("cache算子　　")
//    　　　　14、persist算子
    println("persist算子")
//
//
//    2）Key-Value数据类型的Transfromation算子
//
//    　　一、输入分区与输出分区一对一
//
//    　　　　15、mapValues算子
    println("mapValues算子")
//    　　二、对单个RDD或两个RDD聚集
//
//    　　　单个RDD聚集
//
//    　　　　16、combineByKey算子
    println("combineByKey算子")
//    　　　　17、reduceByKey算子
    println("reduceByKey算子")
//    　　　　18、partitionBy算子
    println("partitionBy算子")
//    　　两个RDD聚集
//
//    　　　　19、Cogroup算子
    println("Cogroup算子")
//    　　三、连接
//
//    　　　　20、join算子
    println("join算子")
//    　　　　21、leftOutJoin和 rightOutJoin算子
    println("leftOutJoin和 rightOutJoin算子")
//
//
//
//
//    3）Action算子
//
//    　　一、无输出
//
//    　　　　22、foreach算子
    println("foreach算子")
//    　　二、HDFS
//
//    　　　　23、saveAsTextFile算子
    println("saveAsTextFile算子")
//    　　　　24、saveAsObjectFile算子
    println("saveAsObjectFile算子")
//    　　三、Scala集合和数据类型
//
//    　　　　25、collect算子
    println("collect算子")
//    　　　　26、collectAsMap算子
    println("collectAsMap算子")
//    　　　  27、reduceByKeyLocally算子
    println("reduceByKeyLocally算子")
//    　　　  28、lookup算子
    println("lookup算子子")
//    　　　　29、count算子
    println("count算子")
//    　　　　30、top算子
    println("top算子")
//    　　　　31、reduce算子
    println("reduce算子")
//    　　　　32、fold算子
    println("fold算子")
//    　　　　33、aggregate算子
    println("aggregate算子")
//
//





    spark.stop()

  }
}
